package cn.iocoder.yudao.module.vos.service.ecdr.admin;

import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
import cn.iocoder.yudao.module.vos.controller.admin.ecdr.vo.ECdrPageReqVO;
import cn.iocoder.yudao.module.vos.controller.admin.ecdr.vo.ECdrSaveReqVO;
import cn.iocoder.yudao.module.vos.dal.dataobject.ecdr.ECdrDO;
import cn.iocoder.yudao.module.vos.dal.dataobject.ecdr.ECdrDOWithoutTag;
import cn.iocoder.yudao.module.vos.dal.mysql.ecdr.ECdrMapper;
import cn.iocoder.yudao.module.vos.job.vo.CdrMongoDO;
import cn.iocoder.yudao.module.vos.utils.CdrUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.BulkWriteOptions;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import lombok.extern.slf4j.Slf4j;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;

import javax.annotation.Resource;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.module.vos.enums.ErrorCodeConstants.E_CDR_NOT_EXISTS;

@Service
@Slf4j
@Validated
public class ECdrServiceImpl implements ECdrService {

    @Resource
    private ECdrMapper eCdrMapper;

    @Autowired
    private MongoTemplate mongoTemplate;

    @Value("${spring.datasource.dynamic.datasource.mysql2.name}")
    private String database2Name;

    @Override
    public Long createECdr(ECdrSaveReqVO createReqVO) {
        ECdrDO eCdr = BeanUtils.toBean(createReqVO, ECdrDO.class);
        eCdrMapper.insert(eCdr);
        return eCdr.getFlowno();
    }

    @Override
    public void updateECdr(ECdrSaveReqVO updateReqVO) {
        validateECdrExists(updateReqVO.getFlowno());
        ECdrDO updateObj = BeanUtils.toBean(updateReqVO, ECdrDO.class);
        eCdrMapper.updateById(updateObj);
    }

    @Override
    public void deleteECdr(Long id) {
        validateECdrExists(id);
        eCdrMapper.deleteById(id);
    }

    @Override
    public void deleteECdrListByIds(List<Long> ids) {
        eCdrMapper.deleteByIds(ids);
    }

    private void validateECdrExists(Long id) {
        if (eCdrMapper.selectById(id) == null) {
            throw exception(E_CDR_NOT_EXISTS);
        }
    }

    @Override
    public ECdrDO getECdr(Long id) {
        return eCdrMapper.selectById(id);
    }

    @Override
    public PageResult<ECdrDO> getECdrPage(ECdrPageReqVO pageReqVO) {
        PageResult<ECdrDO> page = eCdrMapper.selectPage(pageReqVO);
        if (page != null && page.getList() != null) {
            for (ECdrDO row : page.getList()) {
                row.setPhoneTag(CdrUtil.computePhoneTag(row));
            }
        }
        return page;
    }

    @Override
    public List<ECdrDO> getECdrListForExport(ECdrPageReqVO reqVO) {
        List<ECdrDO> list = eCdrMapper.selectAllByShards(reqVO);
        if (list != null) {
            for (ECdrDO row : list) {
                row.setPhoneTag(CdrUtil.computePhoneTag(row));
            }
        }
        return list;
    }

    @Override
    public Long cleanupOldData() {
        return 0L;
    }

    /**
     * 同步指定天数内（可设置截止日）的 CDR 到 Mongo。
     * 整表查出 → 按 calleetogatewaye164 分组 → 每个 phone 单独 upsert。
     */
    @Override
    @DS("mysql2")
    public Long sync(String daysAndDate) {
        String[] param = daysAndDate.split(",");
        int days = Integer.parseInt(param[0]);
        String cutoffDate = (param.length > 1) ? param[1] : null;

        if (days < 1) {
            throw new IllegalArgumentException("Days must be greater than 0");
        }

        AtomicLong totalPhonesUpserted = new AtomicLong(0);
        LocalDate today = LocalDate.now();
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyyMMdd");

        MongoCollection<Document> collection = mongoTemplate.getDb().getCollection("cdr");
        BulkWriteOptions unordered = new BulkWriteOptions().ordered(false);

        for (int i = 1; i <= days; i++) {
            LocalDate target = today.minusDays(i);
            String dayStr = target.format(df);
            String tableName = "e_cdr_" + dayStr;

            if (cutoffDate != null && dayStr.compareTo(cutoffDate) > 0) {
                log.info("跳过 {}（晚于 cutoff {}）", dayStr, cutoffDate);
                continue;
            }

            Integer exists = eCdrMapper.checkTableExists(database2Name, tableName);
            if (!Objects.equals(exists, 1)) {
                log.warn("表不存在：{}", tableName);
                continue;
            }

            log.info("开始同步表：{}", tableName);

            long dayPhoneUpserts = 0L;

            try {
                // ① 一次性查出当日所有需要的列
                List<CdrMongoDO> rows = eCdrMapper.selectAllByDay(tableName);
                if (rows == null || rows.isEmpty()) {
                    log.info("[{}] 无记录可同步", tableName);
                } else {
                    // ② 按号码分组
                    Map<String, List<Document>> grouped = new LinkedHashMap<>();
                    for (CdrMongoDO r : rows) {
                        String phone = CdrUtil.trimNumber(r.getCalleetogatewaye164());
                        if (phone == null || phone.isEmpty()) continue;
                        grouped.computeIfAbsent(phone, k -> new ArrayList<>(8))
                                .add(toCallDoc(r));
                    }

                    // ③ 每个号码一条 upsert（$set records.<day> = calls），不做批量累计
                    for (Map.Entry<String, List<Document>> e : grouped.entrySet()) {
                        String phone = e.getKey();
                        if (phone == null || phone.isEmpty()) continue;
                        List<Document> calls = e.getValue();
                        // 单次 bulkWrite（1 个 UpdateOneModel），或可替换为 collection.updateOne(...) 等价写法
                        collection.bulkWrite(
                                Collections.singletonList(buildSetOp(phone, calls, dayStr)),
                                unordered
                        );
                        dayPhoneUpserts++;
                    }
                }
            } catch (Exception ex) {
                log.error("整表读取/分组/写入 {} 失败", tableName, ex);
            }
            totalPhonesUpserted.addAndGet(dayPhoneUpserts);
            log.info("结束同步表：{}，本日 upsert 手机号数：{}", tableName, dayPhoneUpserts);
            // 轻量休眠，给 GC 一点缓冲
            try {
                System.gc();
                Thread.sleep(1000);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }

        log.info("同步完成，总 upsert 手机号数：{}", totalPhonesUpserted.get());
        return totalPhonesUpserted.get();
    }

    @Override
    public List<ECdrDOWithoutTag> listLastDayCdrByHoldTime(int holdTime, Integer limit, String appointedDate) {
        LocalDate base;
        if (appointedDate != null && !appointedDate.isEmpty()) {
            try {
                base = LocalDate.parse(appointedDate, DateTimeFormatter.ofPattern("yyyyMMdd"));
            } catch (Exception e) {
                base = LocalDate.now();
                log.warn("appointedDate 非法({})，按今天计算", appointedDate);
            }
        } else {
            base = LocalDate.now();
        }
        LocalDate targetDate = base.minusDays(1);
        String fmtTargetDate = targetDate.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
        String tableName = "e_cdr_" + fmtTargetDate;

        Integer exist = eCdrMapper.checkTableExists(database2Name, tableName);
        if (exist == null || !exist.equals(1)) {
            log.error("{} does not exist", tableName);
            return new ArrayList<>();
        }

        // limit 为 null 或 <=0 时，Mapper 层动态 SQL 不加 LIMIT
        return eCdrMapper.listLastDayCdrByHoldTime(holdTime, tableName,
                (limit != null && limit > 0) ? limit : null);
    }


    /** 将一条行记录转换为 Mongo 中的通话明细 Document */
    private static Document toCallDoc(CdrMongoDO r) {
        return new Document("callergatewayid", r.getCallergatewayid())
                .append("calleetogatewaye164", r.getCalleetogatewaye164())
                .append("calleeaccesse164", r.getCalleeaccesse164())
                .append("starttime", r.getStarttime())
                .append("stoptime", r.getStoptime())
                .append("holdtime", r.getHoldtime())
                .append("enddirection", r.getEnddirection());
    }

    /**
     * 构造对单个 phone 的 upsert（$set records.<day>=calls）
     * 对 calls 做深拷贝快照，避免后续清空引用导致写入空数据
     */
    private WriteModel<Document> buildSetOp(String phone, List<Document> calls, String dayKey) {
        if (calls != null) {
            for (Document call : calls) {
                Object start = (call == null) ? null : call.get("starttime");
                log.info("[flushToMongo] day={} phone={} starttime={}", dayKey, phone, start);
            }
        }

        final List<Document> snapshot =
                (calls == null || calls.isEmpty())
                        ? Collections.emptyList()
                        : calls.stream().map(Document::new).collect(Collectors.toList());

        String md5 = CdrUtil.md5(phone);
        String md5p63 = CdrUtil.md5("+63" + phone);

        Document setDoc = new Document("records." + dayKey, snapshot)
                .append("phone_md5", md5)
                .append("phone_md5+63", md5p63);

        return new UpdateOneModel<>(
                Filters.eq("_id", phone),
                new Document("$set", setDoc),
                new UpdateOptions().upsert(true)
        );
    }
}
